package com.mytijian.task.repository.redis;

import com.google.common.collect.Lists;
import com.mytijian.base.exception.BizException;
import com.mytijian.task.enums.SubTaskStatusEnum;
import com.mytijian.task.enums.TaskStatusEnum;
import com.mytijian.task.exception.MtjkTaskProcessException;
import com.mytijian.task.lifecycle.ApplicationLifycycle;
import com.mytijian.task.repository.RuningTaskRepository;
import com.mytijian.task.repository.TaskRepository;
import com.mytijian.task.repository.model.TaskIdentifyDTO;
import com.mytijian.task.task.listener.RuningTaskListener;
import com.mytijian.task.util.IpAddressUtil;
import com.mytijian.task.zookeeper.LeaderService;
import com.mytijian.task.model.TaskIdentify;
import com.mytijian.task.model.TaskValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.lang.Nullable;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * ip1
 *              taskIdentify1 taskValue
 *              taskIdentify2 taskValue
 *
 */
public class RuningTaskRedisRepository extends ApplicationLifycycle implements RuningTaskRepository {
    private Logger logger = LoggerFactory.getLogger(this.getClass());
    private  RedisTemplate redisTemplate;
    private Map<RuningTaskListener, Set<TaskIdentify>> runingTaskListenerSetMap = new HashMap<>();
    private  ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(3, Executors.defaultThreadFactory());
    private LeaderService leaderService;
    private TaskRepository taskRepository;
    private RedisConfig redisConfig;
    public RuningTaskRedisRepository(LeaderService leaderService, TaskRepository taskRepository, RedisConfig redisConfig) {
        this.leaderService = leaderService;
        this.taskRepository = taskRepository;
        this.redisConfig = redisConfig;
    }

   public void addRunTaskListener(RuningTaskListener runingTaskListener) {
       runingTaskListenerSetMap.computeIfAbsent(runingTaskListener, k -> new HashSet<>(1024));
   }
    @Override
    public void init() {
        initRedisTemplate();
        initScheduled();
    }

    @Override
    public int initSort() {
        return 2;
    }

    @Override
    public void destory() {
        scheduledExecutorService.shutdownNow();
        runingTaskListenerSetMap = null;
    }

    @Override
    public int destorySort() {
        return 1;
    }

    private String split = ":";
    // 任务运行中的节点
    private String getTaskNameSpace(String ip) {
        return "mtjk-task" + split + "runging" + split + ip + split;
    }
    private  String getTaskIdNameSpace() {
        return "mtjk-task" + split + "taskId";
    }
    private void initRedisTemplate() {
        RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration();
        redisStandaloneConfiguration.setHostName(redisConfig.getHostName());
        redisStandaloneConfiguration.setPassword(redisConfig.getPassword());
        redisStandaloneConfiguration.setPort(redisConfig.getPort());
        LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory(redisStandaloneConfiguration);
        lettuceConnectionFactory.afterPropertiesSet();
        redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(lettuceConnectionFactory);
        redisTemplate.setKeySerializer(RedisSerializer.string());
        redisTemplate.setValueSerializer(RedisSerializer.string());
        redisTemplate.setHashKeySerializer(RedisSerializer.string());
        redisTemplate.setHashValueSerializer(RedisSerializer.string());
        redisTemplate.setDefaultSerializer(RedisSerializer.string());
        redisTemplate.afterPropertiesSet();
    }
    private void initScheduled() {
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    Map<TaskIdentify, TaskValue> remoteTaskIdentify = getByIp(IpAddressUtil.getLocalHost(), 10000);
                    Map<RuningTaskListener, Set<TaskIdentify>> notNeedUpdateTaskIdentifyMap = new HashMap<>();
                    runingTaskListenerSetMap.forEach(((runingTaskListener, taskIdentifies) -> {
                        notNeedUpdateTaskIdentifyMap.put(runingTaskListener, new HashSet<>(1024));
                    }));
                    remoteTaskIdentify.forEach(((taskIdentify, taskValue) -> {
                        Set<Map.Entry<RuningTaskListener,Set<TaskIdentify>>> entries = runingTaskListenerSetMap.entrySet();
                        for (Map.Entry<RuningTaskListener, Set<TaskIdentify>> entry : entries) {
                            // 本地缓存不包含
                            if (!entry.getValue().contains(taskIdentify)) {
                                try {
                                    // 通知远程新增任务
                                    boolean needUpdateLocalCache = entry.getKey().newTaskIdentify(taskIdentify, taskValue);
                                    if (!needUpdateLocalCache) {
                                        notNeedUpdateTaskIdentifyMap.get(entry.getKey()).add(taskIdentify);
                                    }
                                }catch (Exception e) {
                                    logger.error("新任务通知失败 runTaskListener = {} taskIdentify= {} e= {}", entry.getKey().getClass().getName(), taskIdentify.getTaskKey(), e);
                                }
                            }
                        }
                    }));
                    Set<TaskIdentify> remoteTaskIdentifyKes = remoteTaskIdentify.keySet();
                    runingTaskListenerSetMap.forEach(((runingTaskListener, taskIdentifies) -> {
                        // 不需要更新本地缓存的集合
                        Set<TaskIdentify> notNeedUpdateTaskIdentifies = notNeedUpdateTaskIdentifyMap.get(runingTaskListener);
                        taskIdentifies.clear();
                        for (TaskIdentify taskIdentify : remoteTaskIdentifyKes) {
                            if (!notNeedUpdateTaskIdentifies.contains(taskIdentify)) {
                                taskIdentifies.add(taskIdentify);
                            }
                        }

                    }));
                }catch (Exception e) {
                    logger.error("同步远程任务到内存出错 e= {}", e);
                }

            }
        }, 5,5, TimeUnit.SECONDS);
//        // 远程任务更新到mysql
//        // 失败的信息已经实时写入
//        //这里写成功的信息
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    Map<TaskIdentify, TaskValue> taskValueMap = getByIp(IpAddressUtil.getLocalHost(), 10000);
                    taskValueMap.forEach(((taskIdentify, taskValue) -> {
                        List<Integer> subTaskIds  = taskRepository.getSubTaskIds(taskIdentify.getTaskId(), taskIdentify.getBeginSubTaskId(), taskIdentify.getEndSubTaskId(), Lists.newArrayList(SubTaskStatusEnum.un_begin.getCode()));

                        if (!CollectionUtils.isEmpty(subTaskIds) && taskValue.getCompleteSubTaskId() != null) {
                            List<Integer> processSuccessSubTaskIds = new ArrayList<>(256);
                            for (Integer subTaskId : subTaskIds) {
                                if (subTaskId <= taskValue.getCompleteSubTaskId()) {
                                    processSuccessSubTaskIds.add(subTaskId);
                                }
                            }
                            if (!CollectionUtils.isEmpty(processSuccessSubTaskIds)) {
                                taskRepository.processSuccess(taskIdentify.getTaskId(), processSuccessSubTaskIds);
                            }

                        }
                        // 更新完成 && taskIdentify 到终态
                        if (TaskStatusEnum.getEndStatusInteger().contains(taskValue.getTaskStuts())) {
                            taskRepository.save(taskIdentify, taskValue, IpAddressUtil.getLocalHost());
                            // 在这里宕机也不要紧，机器重启的时候依然可以删除
                            remove(Lists.newArrayList(taskIdentify), IpAddressUtil.getLocalHost());
                            removeRemoteTaskIdentify(taskIdentify);
                            logger.info("保存完成状态的 taskIdentify = {}, ip = {}", taskIdentify.getTaskKey(), IpAddressUtil.getLocalHost());
                        }
                    }));
                }catch (Exception e) {
                    logger.error("远程任务更新到mysql 失败 e= {}",e);
                }
            }
        }, 5,5, TimeUnit.SECONDS);

        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    if (leaderService.isLeader()) {
                        Map<String, String> taskIdMap = redisTemplate.opsForHash().entries(getTaskIdNameSpace());
                        taskIdMap.forEach((taskId, value) ->{
                            List<TaskIdentifyDTO> taskIdentifyDTOS = taskRepository.getTaskIdentifies(Integer.parseInt(taskId));
                            List<TaskIdentifyDTO> tempList = taskIdentifyDTOS.stream().filter(taskIdentifyDTO -> !TaskStatusEnum.getEndStatusInteger().contains(taskIdentifyDTO.getTaskStatus())).collect(Collectors.toList());
                            //所有的分配任务都到达终态
                            if (CollectionUtils.isEmpty(tempList)) {
                                taskRepository.completeStatus(taskIdentifyDTOS);
                                //在这里宕机也失败也不要紧，下一次进入的时候也可以删除
                                redisTemplate.opsForHash().delete(getTaskIdNameSpace(), taskId);
                            }
                        });
                    }
                }catch (Exception e) {
                    logger.error("删除已经到终态的任务 失败 e={}",e);
                }
            }
        }, 5,5, TimeUnit.SECONDS);
    }
    private void removeRemoteTaskIdentify(TaskIdentify taskIdentify) {
        if (CollectionUtils.isEmpty(runingTaskListenerSetMap)) {
            return;
        }
        runingTaskListenerSetMap.forEach(((runingTaskListener, taskIdentifies) -> {
            try {
                runingTaskListener.removeRemoteTaskIdentify(taskIdentify);
            }catch (Exception e) {
                logger.error("远程任务删除通知失败 runingTaskListener = {}, taskIdentify = {} , e= {}", runingTaskListener.getClass().getSimpleName(), taskIdentify.getTaskKey(), e);
            }
        }));
    }
    public Map<TaskIdentify, TaskValue> getByIp(String ip, Integer limit) {
        Map<TaskIdentify, TaskValue> map = new HashMap<>();
        ScanOptions scanOptions = ScanOptions.scanOptions().count(limit).build();
        try(Cursor<Map.Entry<String, String>> cursor =  redisTemplate.opsForHash().scan(getTaskNameSpace(ip), scanOptions)) {
            while (cursor.hasNext()) {
                Map.Entry<String, String> entry = cursor.next();
                TaskIdentify taskIdentify = TaskIdentify.build(entry.getKey());
                TaskValue taskValue = TaskValue.build(entry.getValue());
                map.put(taskIdentify, taskValue);
            }
        }catch (Exception e) {
            logger.error("查询失败 e = {}", e);
        }
        return map;
    }

    @Override
    public Map<TaskIdentify, TaskValue> getByIp(String ip) {
        Map<TaskIdentify, TaskValue> map = new HashMap<>();
        try {
            redisTemplate.execute(new RedisCallback<String>() {
                @Nullable
                @Override
                public String doInRedis(RedisConnection redisConnection) throws DataAccessException {
                    // taskId taskIdentify1=taskValue1
                    Map<byte[], byte[]> bytesMap = redisConnection.hGetAll(serialize(getTaskNameSpace(ip)));

                    if (CollectionUtils.isEmpty(bytesMap)) {
                        return "ok";
                    }
                    bytesMap.forEach((taskIdentifyByte, taskValueByte) ->{
                        map.put(toTaskIdentify(taskIdentifyByte), toTaskValue(taskValueByte));
                    });

                    return "ok";
                }
            });
        }catch (Exception e) {
            logger.error("获取任务列表失败 ip = {} e= {}", ip, e);
            throw new MtjkTaskProcessException("12", "获取任务列表失败");
        }

        return map;
    }

    @Override
    public Map<TaskIdentify, TaskValue> getByIpAndTaskId(String ip, Integer taskId) {
        Map<TaskIdentify, TaskValue> map = new HashMap<>();

        redisTemplate.execute(new RedisCallback() {
            @Nullable
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException {
                Map<byte[], byte[]> bytesMap = connection.hGetAll(serialize(getTaskNameSpace(ip)));
                bytesMap.forEach((taskIdentifyByte, taskValueByte)-> {
                    TaskIdentify taskIdentify = toTaskIdentify(taskIdentifyByte);
                    if (taskIdentify.getTaskId().equals(taskId)) {
                        map.put(taskIdentify, toTaskValue(taskValueByte));
                    }
                });
                return "ok";
            }
        });
        return map;
    }

    @Override
    public TaskValue getByIdentify(String ip, TaskIdentify taskIdentify) {
        final TaskValue[] taskValue = {null};
        try {
            redisTemplate.execute(new RedisCallback() {
                @Nullable
                @Override
                public Object doInRedis(RedisConnection connection) throws DataAccessException {
                    byte[] bytes = connection.hGet(serialize(getTaskNameSpace(ip)), serialize(taskIdentify.getTaskKey()));
                    if (bytes == null) {
                        return "ok";
                    }
                    taskValue[0] = toTaskValue(bytes);
                    return "ok";
                }
            });
        }catch (Exception e ) {
            logger.error("获取redis 任务失败 ip= {}, taskIdentify = {}", ip, taskIdentify.getTaskKey());
            throw new MtjkTaskProcessException("12", "获取taskValue失败 ");
        }

        return taskValue[0];
    }
    @Override
    public void transfer(Map<TaskIdentify, TaskValue> map, String sourceIp, String destIp) {
        if (CollectionUtils.isEmpty(map)) {
            return;
        }
        try {
            redisTemplate.execute(new SessionCallback() {
                @Nullable
                @Override
                public Object execute(RedisOperations operations) throws DataAccessException {
                    operations.multi();
                    // 在事务中执行
                    remove(map.keySet(), sourceIp);
                    map.forEach(((taskIdentify, taskValue) -> {
                        save(destIp, taskIdentify, taskValue);
                    }));
                    return operations.exec();
                }
            });
        }catch (Exception e ) {
            logger.error("迁移任务错误 map = {}, sourceIp = {}, destIp ={} e= {}", map, sourceIp, destIp, e);
            throw new MtjkTaskProcessException("12", "迁移任务错误");
        }




    }


    private void remove(Collection<TaskIdentify> taskIdentifies, String ip) {
        List<String> taskIdentifyString = new ArrayList<>(taskIdentifies.size());
        for (TaskIdentify taskIdentify : taskIdentifies) {
            taskIdentifyString.add(taskIdentify.getTaskKey());
        }
        redisTemplate.opsForHash().delete(getTaskNameSpace(ip), taskIdentifyString.toArray(new String[]{}));

    }
    @Override
    public void save(TaskIdentify taskIdentify, TaskValue taskValue, String ip) {
        try {
            save(ip, taskIdentify, taskValue);
        }catch (Exception e) {
            logger.error("redis保存异常 taskIdentify = {}, taskValue = {}, ip = {} e= {}", taskIdentify.getTaskKey(), taskValue.getValue(), ip, e);
            throw new MtjkTaskProcessException("12","redis保存异常");
        }

    }

//    private List<TaskIdentify> toTaskIdentifies(String listTaskIdentifies) {
//        listTaskIdentifies = listTaskIdentifies.replace("[", "");
//        listTaskIdentifies = listTaskIdentifies.replace("]", "");
//        List<TaskIdentify> taskIdentifies = new ArrayList<>();
//        for (String taskIdentifyString :listTaskIdentifies.split(",")) {
//            taskIdentifies.add(TaskIdentify.build(taskIdentifyString));
//        }
//        return taskIdentifies;
//    }

    private void save(String ip, TaskIdentify taskIdentify, TaskValue taskValue) {
        redisTemplate.opsForHash().put(getTaskNameSpace(ip), taskIdentify.getTaskKey(), taskValue.getValue());
        redisTemplate.opsForHash().put(getTaskIdNameSpace(), taskIdentify.getTaskId() +"","" );
    }

//    public static void main(String[] args) {
//        List<TaskIdentify> taskIdentifies = new ArrayList<>();
//        taskIdentifies.add(new TaskIdentify(1, 1, 1));
//        taskIdentifies.add(new TaskIdentify(2, 2, 2));
//        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
//        byte[] bytes = stringRedisSerializer.serialize(taskIdentifies.toString());
//        String s = stringRedisSerializer.deserialize(bytes);
//        System.out.println(s);
//    }
    private String deserialize(byte[] bytes) {
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        return stringRedisSerializer.deserialize(bytes);
    }

    private TaskIdentify toTaskIdentify(byte[] bytes) {
        String key = deserialize(bytes);
        return TaskIdentify.build(key);
    }
    private TaskValue toTaskValue(byte[] bytes) {
        String value = deserialize(bytes);
        return TaskValue.build(value);
    }
    private byte[] serialize(String string) {
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        return stringRedisSerializer.serialize(string);
    }

}
